package com.example.mq;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.springframework.stereotype.Component;

/**
 * RocketMqConsumerService
 *
 * 提供启动消费者的方法，使用 DefaultMQPushConsumer 批量拉取消息。
 */
@Component
public class RocketMqConsumerService {

    private final String namesrvAddr = "localhost:9876";

    /**
     * startConsumer：启动消费者并订阅指定主题
     * @param consumerGroup 消费者组名称
     * @param topic 订阅主题
     * @param listener 消息监听器（批量消费）
     * @return 启动后的消费者实例
     */
    public DefaultMQPushConsumer startConsumer(String consumerGroup, String topic, MessageListenerConcurrently listener) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.subscribe(topic, "*");
        // 设置每批拉取消息数量，例：10条
        consumer.setConsumeMessageBatchMaxSize(10);
        consumer.registerMessageListener(listener);
        consumer.start();
        System.out.println("RocketMqConsumerService 启动成功，消费组：" + consumerGroup + "，主题：" + topic);
        return consumer;
    }
}